package com.example.demo.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.SeekToCurrentBatchErrorHandler;
import org.springframework.retry.RetryPolicy;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.backoff.FixedBackOff;

@Configuration
public class KafkaConfig {

    /**
     * 配置Kafka监听器容器工厂，实现固定1秒间隔的无限重试
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory) {
        
        ConcurrentKafkaListenerContainerFactory<String, String> factory = 
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        
        // 设置确认模式为手动立即确认
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        
        // 配置重试模板 - 固定1秒间隔的无限重试
        RetryTemplate retryTemplate = new RetryTemplate();
        
        // 设置重试策略：无限重试
        RetryPolicy retryPolicy = new SimpleRetryPolicy(Integer.MAX_VALUE);
        retryTemplate.setRetryPolicy(retryPolicy);
        
        // 设置固定退避策略：每次重试间隔1秒
        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        backOffPolicy.setBackOffPeriod(1000L); // 1秒 = 1000毫秒
        retryTemplate.setBackOffPolicy(backOffPolicy);
        
        factory.setRetryTemplate(retryTemplate);
        
        // 当重试耗尽时的恢复回调（理论上不会被调用，因为是无限重试）
        factory.setRecoveryCallback(context -> {
            System.err.println("重试耗尽（理论上不应该到达这里）: " + context.getLastThrowable().getMessage());
            return null;
        });
        
        return factory;
    }

    /**
     * 配置Kafka批量监听器容器工厂，支持批量消息处理
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> batchKafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory) {
        
        ConcurrentKafkaListenerContainerFactory<String, String> factory = 
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        
        // 启用批量监听模式
        factory.setBatchListener(true);
        
        // 设置确认模式为手动立即确认
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        
        // 配置批量消费参数
        ContainerProperties containerProperties = factory.getContainerProperties();
        containerProperties.setPollTimeout(1000); // 1秒超时
        
        // 配置批量错误处理器 - 替代RetryTemplate
        SeekToCurrentBatchErrorHandler errorHandler = new SeekToCurrentBatchErrorHandler();
        
        // 配置重试策略：固定间隔1秒，最多重试Integer.MAX_VALUE次（无限重试）
        FixedBackOff backOff = new FixedBackOff(1000L, Long.MAX_VALUE); // 1秒间隔，无限重试
        errorHandler.setBackOff(backOff);
        
        // 设置错误处理器
        factory.setBatchErrorHandler(errorHandler);
        
        return factory;
    }

    /**
     * 配置带有自定义错误处理器的批量监听器容器工厂
     * 更好地与KafkaOffsetSkipAspect配合工作
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> customBatchKafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory) {
        
        ConcurrentKafkaListenerContainerFactory<String, String> factory = 
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        
        // 启用批量监听模式
        factory.setBatchListener(true);
        
        // 设置确认模式为手动立即确认
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        
        // 配置批量消费参数
        ContainerProperties containerProperties = factory.getContainerProperties();
        containerProperties.setPollTimeout(1000); // 1秒超时
        
        // 使用自定义错误处理器，直接传递异常给KafkaOffsetSkipAspect
        factory.setBatchErrorHandler(new CustomBatchErrorHandler());
        
        return factory;
    }
}
